package com.shekhar.paas.archiver;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import twitter4j.Tweet;

import com.shekhar.paas.domain.ArchivingJobLogBuilder;
import com.shekhar.paas.domain.Story;
import com.shekhar.paas.exceptions.StoryCreationException;
import com.shekhar.paas.search.SearchTerms;
import com.shekhar.paas.story.SimilarStoryFinder;
import com.shekhar.paas.story.StoryCreator;
import com.shekhar.paas.story.StoryScorer;
import com.shekhar.paas.story.TweetBuilder;
import com.shekhar.paas.twitter.TwitterService;
import com.shekhar.paas.twitter.utils.TwitterUtils;

/**
 * Responsibilities of TweetsArchiver <li>Search for last 100 tweets about a
 * topic.A topic can have multiple search terms for example "cloud foundry" has
 * "cfoundry","cloudfoundry etc."</li> <li>Filter out all the tweets which we
 * have already processed. As we are polling twitter after a scheduled period we
 * might get same stories and we would not like to process them again</li> <li>
 * then for each tweet we try to create a story. A story will be created for
 * only those tweets which have links in them.</li> <li>Now for each story
 * corresponding to a tweet(Yes a single tweet can have multiple stories) we
 * find out which all stories already exist in datastore.</li> <li>We don't want
 * to create new stories for the links which already have stories.</li> <li>All
 * the new stories are directly persisted in MongoDB</li> <li>For all the
 * already persisted stories we find whether it a retweet or a new tweet and
 * update the story and its score</li>
 * 
 * @author shekhar
 * 
 */
@Component
public class TweetsArchiver {

    private Logger logger = Logger.getLogger(this.getClass());

    @Autowired
    private TwitterService twitterService;

    @Autowired
    private StoryCreator tweetStoryCreator;

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private SimilarStoryFinder similarStoryFinder;

    @Scheduled(fixedRate = 1800000l)
    public void archiveTweets() {
        Date startedAt = new Date();
        ArchivingJobLogBuilder builder = ArchivingJobLogBuilder.archivingJobLog();
        builder.withStartedAt(startedAt);
        long startTime = System.currentTimeMillis();
        try {
            for (String searchTerm : SearchTerms.PAAS_TERMS) {
                try {
                    List<Tweet> tweets = twitterService.search(searchTerm);
                    List<Tweet> filteredTweets = filterOutAlreadyPersistedTweets(tweets);
                    builder.withMessage("For searchTerm " + searchTerm + " , found " + filteredTweets.size()
                            + " number of tweets for which we are generating stories.");
                    builder.withNumberOfTweetsProcessed(builder.numberOfTweetProcessed() + filteredTweets.size());
                    for (Tweet tweet : filteredTweets) {
                        List<Story> stories = null;
                        try {
                            stories = tweetStoryCreator.createNewStory(tweet, searchTerm);
                        } catch (StoryCreationException e) {
                            builder.withTweetsNotProcessed(e.getText());
                            continue;
                        }

                        List<Story> alreadyPersistedStories = filterAlreadyPersistedStories(stories);
                        if (CollectionUtils.isEmpty(alreadyPersistedStories)) {
                            for (Story story : stories) {
                                similarStoryFinder.associateSimilarStoriesTogether(story);
                            }
                            mongoTemplate.insertAll(stories);
                        } else if (!CollectionUtils.isEmpty(alreadyPersistedStories) && stories.size() > alreadyPersistedStories.size()) {
                            stories.removeAll(alreadyPersistedStories);
                            for (Story story : stories) {
                                similarStoryFinder.associateSimilarStoriesTogether(story);
                            }
                            mongoTemplate.insertAll(stories);
                        }

                        for (Story alreadyPersistedStory : alreadyPersistedStories) {
                            Query query = Query.query(Criteria.where("id").is(alreadyPersistedStory.getId()));
                            if (TwitterUtils.isRetweet(tweet.getText())) {
                                Story updatedStory = mongoTemplate.findAndModify(query,
                                        new Update().inc("retweetCount", 1).set("updatedOn", startedAt), new FindAndModifyOptions().returnNew(true),
                                        Story.class, "stories");
                                long score = new StoryScorer().score(updatedStory);

                                mongoTemplate.updateFirst(query, Update.update("score", score), "stories");

                            } else {
                                Story updatedStory = mongoTemplate.findAndModify(query,
                                        new Update().inc("tweetCount", 1).set("updatedOn", startedAt).push("tweets", TweetBuilder.tweet()
                                                .withCreatedAt(tweet.getCreatedAt())
                                                .withFromUser(tweet.getFromUser())
                                                .withId(tweet.getId())
                                                .withProfileImageUrl(tweet.getProfileImageUrl())
                                                .withText(tweet.getText())
                                                .withIsoLanguageCode(tweet.getIsoLanguageCode())
                                                .build()), new FindAndModifyOptions().returnNew(true),
                                        Story.class, "stories");

                                long score = new StoryScorer().score(updatedStory);
                                mongoTemplate.updateFirst(query, Update.update("score", score), "stories");
                            }
                        }

                    }

                } catch (Exception e) {
                    logger.error("Exception while running archiving job ....", e);
                    builder.withMessage("Exception while running archiving job for search term " + searchTerm);
                    builder.withMessage(ExceptionUtils.getStackTrace(e));
                }

            }

        } catch (Exception e) {
            logger.error("Exception while running archiving job ....", e);
            builder.withMessage(ExceptionUtils.getStackTrace(e));
        } finally {
            builder.withFinishedAt(new Date());
            long endTime = System.currentTimeMillis();
            long numberOfSeconds = (endTime - startTime) / 1000;
            builder.withTimeTaken(numberOfSeconds);
            mongoTemplate.insert(builder.build());
        }

    }

    private List<Story> filterAlreadyPersistedStories(List<Story> stories) {
        List<Story> alreadyPersistedStories = new ArrayList<Story>();
        for (Story story : stories) {
            String storyUrl = story.getUrl();
            if (!StringUtils.endsWith(storyUrl, "/")) {
                storyUrl += "/";
            }
            Query queryForUrl = Query.query(Criteria.where("url").is(storyUrl));
            Story alreadyPersistedStory = mongoTemplate.findOne(queryForUrl, Story.class, "stories");
            if (alreadyPersistedStory != null) {
                alreadyPersistedStories.add(alreadyPersistedStory);
                continue;
            }
            Query queryForTitle = Query.query(Criteria.where("title").is(story.getTitle()));
            alreadyPersistedStory = mongoTemplate.findOne(queryForTitle, Story.class, "stories");
            if (alreadyPersistedStory != null) {
                alreadyPersistedStories.add(alreadyPersistedStory);
            }
        }
        return alreadyPersistedStories;
    }

    private List<Tweet> filterOutAlreadyPersistedTweets(List<Tweet> tweets) {
        List<Tweet> filteredTweets = new ArrayList<Tweet>();
        for (Tweet tweet : tweets) {
            Story persistedStory = mongoTemplate.findOne(Query.query(Criteria.where("tweets._id").is(tweet.getId())), Story.class);
            if (persistedStory != null) {
                continue;
            }
            filteredTweets.add(tweet);
        }
        return filteredTweets;
    }
}
